package com.gjy.disruptor.db;


import com.lmax.disruptor.EventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-08-07 10:45:53
 */
public class DbConsumer implements EventHandler<DbEntity> {
    private static final Logger log = LoggerFactory.getLogger(DbConsumer.class);

    private static final Integer BATCH_SIZE = 5000;
    private static final List<String> sqlList = new CopyOnWriteArrayList<>();
    private final Connection connection;

    public DbConsumer(String url, String username, String password) {
        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection = DriverManager.getConnection(url, username, password);
            connection.setAutoCommit(false);
        } catch (SQLException | ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void onEvent(DbEntity event, long sequence, boolean endOfBatch) throws Exception {
        String sql = tranEntityToSql(event);
        sqlList.add(sql);
        if (endOfBatch || sqlList.size() >= BATCH_SIZE) {
            processSql();
        }
    }

    private void processSql() {
        try (Statement statement = connection.createStatement()) {
            for (String s : sqlList) {
                statement.addBatch(s);
            }
            statement.executeBatch();
            connection.commit();
            statement.clearBatch();
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            log.info("处理了: {}", sqlList.size());
            sqlList.clear();
        }
    }

    private String tranEntityToSql(DbEntity entity) {
        return entity.getSql();
    }

}
